package com.bs.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.bs.common.utils.SnowFlakeUtil;
import com.bs.kafka.vo.UserVo;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;

/**
 * @author bingshao
 * @date 2023/6/9
 **/
@RestController
@RequestMapping("/testProducer")
public class TestProducerController {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    // 下面的代码为测试kafka的功能

    @PostMapping("/send1")
    public void send1() {
        UserVo userVo = new UserVo();
        Long id = SnowFlakeUtil.getId();
        userVo.setId(id);
        userVo.setName("普通发送");
        userVo.setCreateTime(new Date());
        kafkaTemplate.send("test-topic", JSON.toJSONString(userVo));
    }

    @PostMapping("/send2")
    public void send2() {
        UserVo userVo = new UserVo();
        userVo.setId(SnowFlakeUtil.getId());
        userVo.setName("发送Message对象");
        userVo.setCreateTime(new Date());
        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(userVo)).setHeader(KafkaHeaders.TOPIC, "test-topic").build();
        kafkaTemplate.send(message);
    }

    @PostMapping("/send3")
    public void send3() {
        UserVo userVo = new UserVo();
        Long id = SnowFlakeUtil.getId();
        userVo.setId(id);
        userVo.setName("发送带key的对象");
        userVo.setCreateTime(new Date());

        // 为保证消息的可靠性，可以在发送前将信息持久化一下，比如redis，mysql都可以
        redisTemplate.opsForValue().set(id.toString(), JSON.toJSONString(userVo));

        // 可以在发送的时候添加回调，也可以在listener里面统一添加，我们在listener统一添加
        kafkaTemplate.send("test-topic", id.toString(), JSON.toJSONString(userVo));

        // key计算hash值来确认发送到哪个分区上面
//        kafkaTemplate.send("test-topic", id.toString(), JSON.toJSONString(userVo));

        // 如果同时含有分区和key，则已指定的分区为准
//        kafkaTemplate.send("test-topic", 0, id.toString(), JSON.toJSONString(userVo));

        // 可以带个发送消息的时间的参数  server.properties配置文件log.message.timestamp.type的选项（需要手动添加，默认为CreateTime） 1:CreateTime，创建消息的时间,2:LogAppendTime，写入磁盘的时间
//        kafkaTemplate.send("test-topic", 0, System.currentTimeMillis(), id.toString(), JSON.toJSONString(userVo));
    }
}
